package org.danger.t7.common.ws;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;

import javax.websocket.OnClose;
import javax.websocket.OnError;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.ServerEndpoint;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.danger.t7.common.spring.SpringContext;
import org.danger.t7.common.ws.cloud.WsRedisBus;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

/**
 * 通知消息实时推送前端
 */
@Slf4j
@ServerEndpoint(value = "/ws/conn", configurator = FkWebSocketConfigurator.class)
@Component
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class FkWebSocket {

    /**
     * 记录当前在线连接数
     */
    private static AtomicInteger onlineCount = new AtomicInteger(0);

    /**
     * 存放所有在线的客户端
     */
    private static Map<String, Session> onlineClients = new ConcurrentHashMap<>();

    /**
     * 存放与用户绑定的客户端(当同一用户打开多个页面时，需要同时推送到所有连接)
     */
    private static Map<String, List<String>> userClients = new ConcurrentHashMap<>();

    // 怎么wire不进呢
    @Autowired
    private WsEventHandler handler;

    @Autowired(required = false)
    private WsRedisBus wsRedisBus;

    public static int getSessionCount() {
        return onlineCount.get();
    }

    public static int getOnlineUserCount() {
        return userClients.size();
    }

    /**
     * 连接建立成功调用的方法
     *
     * @param session session
     */
    @OnOpen
    public void onOpen(Session session) {
        onlineCount.incrementAndGet(); //在线数加1
        onlineClients.put(session.getId(), session);
        String userId = getUserId(session);
        if (userId != null) {
            //更新同一用户连接的终端信息
            if (userClients.get(userId) == null) {
                List<String> sessionIds = new ArrayList<>();
                sessionIds.add(session.getId());
                userClients.put(userId, sessionIds);
            } else {
                userClients.get(userId).add(session.getId());
            }
            getHandler().onClientConnected(userId, userClients.get(userId));
        } else {
            try {
                session.close();
            } catch (IOException e) {
                log.error("onOpen出错", e);
            }
        }
    }

    /**
     * 连接建立成功调用的方法
     *
     * @param session session
     */
    @OnClose
    public void onClose(Session session) {
        String userId = getUserId(session);
        String sessionId = session.getId();
        closeSession(session, userId);
        getHandler().onClientClosed(userId, sessionId, userClients.get(userId));
    }

    /**
     * 收到客户端消息后调用的方法
     *
     * @param session session
     * @param message message
     */
    @OnMessage
    public void onMessage(Session session, String message) {
        String userId = getUserId(session);
        getHandler().onReceive(userId, session.getId(), message);
    }

    /**
     * 发生错误时调用的方法
     *
     * @param session session
     * @param error   error
     */
    @OnError
    public void onError(Session session, Throwable error) {
        String userId = getUserId(session);
        String sessionId = session.getId();
        closeSession(session, userId);
        getHandler().onClientClosed(userId, sessionId, userClients.get(userId));
        log.error("有连接报错:{}-{}", session.getId(), userId, error);
    }

    public void send(String fromUid, String toUid, String message) {
        if (!StringUtils.hasLength(fromUid)) {
            fromUid = WsRedisBus.BROKER_UID;
        }
        if (wsRedisBus != null) {
            wsRedisBus.sendMessage(fromUid, toUid, message);
        } else {
            FkWebSocket.sendMessageInThisBroker(fromUid, toUid, message);
        }
    }

    /**
     * 向客户端发通知
     *
     * @param toUid   userId
     * @param message message
     */
    public static void sendMessageInThisBroker(String fromUid, String toUid, String message) {
        log.info("发送：from 【{}】 to 【{}】 ----- {}", fromUid, toUid, message);
        if (!StringUtils.hasLength(fromUid)) {
            fromUid = WsRedisBus.BROKER_UID;
        }
        //给用户主动发送消息时，集群部署情况下，因为不知道用户连接的是哪个提供 WebSocket 的应用服务，
        //所以可以通过消息队列广播消费，每个应用判断待通知用户是否是和自己提供的WebSocket服务连接，如果是，则推送消息给用户
        try {
            if (toUid != null && userClients.get(toUid) != null) {
                String text = fromUid + WsRedisBus.SPLITER + message;
                List<String> invalidSessionList = new ArrayList<>();
                for (String sessionId : userClients.get(toUid)) {
                    Session toSession = onlineClients.get(sessionId);
                    if (toSession != null) {
                        Future<Void> sendFuture = toSession.getAsyncRemote().sendText(text);
                        sendFuture.get();
                    } else {
                        //当sessionId在连接中不存在时，记录
                        invalidSessionList.add(sessionId);
                    }
                }
                //清理失效连接
                if (invalidSessionList != null && invalidSessionList.size() > 0) {
                    userClients.get(toUid).removeAll(invalidSessionList);
                }
            } else {
                log.info("未找到有效session：{}", toUid);
            }
        } catch (Exception e) {
            log.error("send fail", e);
        }
    }

    /**
     * 根据seesion获取登录用户ID
     *
     * @param session session
     * @return 用户ID
     */
    private String getUserId(Session session) {
        Object userIdInfo = session.getUserProperties().get("uid");
        String userId = userIdInfo == null ? null : (String) userIdInfo;
        return userId;
    }

    /**
     * 当连接关闭或者连接异常时，清除相关连接信息
     *
     * @param session session
     * @param userId  userId
     */
    private void closeSession(Session session, String userId) {
        onlineCount.decrementAndGet(); //在线数减1
        onlineClients.remove(session.getId());
        if (userId != null && userClients.get(userId) != null) {
            //更新同一用户连接的终端信息
            userClients.get(userId).remove(session.getId());
        }
    }

    private WsEventHandler getHandler(){
        if(handler == null){
            handler = SpringContext.getBean(WsEventHandler.class);
        }
        return handler;
    }
}
